package lly.impl;


import lly.Function;
import lly.Promise;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * Created by lly on 16/4/29.
 */
public class PromiseImpl<T> implements Promise<T> {

    private static final int STATE_PENDING = 0;
    private static final int STATE_FULFILLED = 1;
    private static final int STATE_REJECTED = 2;

    private AtomicInteger state;
    private volatile Object result;

    private BlockingQueue<Function> fulfilledQueue;
    private BlockingQueue<Function> rejectedQueue;

    private Executor executor = Executors.newSingleThreadExecutor();

    public PromiseImpl(BiConsumer<Consumer<T>, Consumer<Exception>> handler) {
        state = new AtomicInteger(STATE_PENDING);
        fulfilledQueue = new LinkedBlockingQueue<>();
        rejectedQueue = new LinkedBlockingQueue<>();
        executor.execute(() -> {handler.accept(this::resolve, this::rejected);});
    }

    private PromiseImpl(BlockingQueue<Function> fulfilledQueue, BlockingQueue<Function> rejectedQeque, Executor executor, BiConsumer<Consumer<T>, Consumer<Exception>> handler) {
        this.fulfilledQueue = fulfilledQueue;
        this.rejectedQueue = rejectedQeque;
        this.executor = executor;
        state = new AtomicInteger(STATE_PENDING);
        executor.execute(() -> {handler.accept(this::resolve, this::rejected);});
    }

    private void resolve(T result) {
        if (!state.compareAndSet(STATE_PENDING, STATE_FULFILLED)) {
            throw new IllegalStateException();
        }
        try {
            @SuppressWarnings("unchecked") Function<T, ?> function = fulfilledQueue.take();
            @SuppressWarnings("unchecked") Function<Exception, ?> error = rejectedQueue.take();
            if (function instanceof Null)
                this.result = null;
            else {
                this.result = function.apply(result);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (Exception e) {
            this.result = e;
        }
    }

    private void rejected(Exception exception) {
        if (!state.compareAndSet(STATE_PENDING, STATE_REJECTED)) {
            throw new IllegalStateException();
        }
        try {
            @SuppressWarnings("unchecked") Function<T, ?> function = fulfilledQueue.take();
            @SuppressWarnings("unchecked") Function<Exception, ?> error = rejectedQueue.take();
            if (error instanceof Null)
                this.result = null;
            else {
                this.result = error.apply(exception);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (Exception e) {
            this.result = e;
        }
    }


    @Override
    public <R> Promise<R> then(Function<T, R> fulfilled, Function<Exception, R> rejected) {
        if (fulfilled == null)
            fulfilledQueue.add(new Null());
        else
            fulfilledQueue.add(fulfilled);

        if (rejected == null)
            rejectedQueue.add(new Null());
        else
            rejectedQueue.add(rejected);

        //可靠性保证: executor为单线程执行器, 下方的lambda必定在result得到结果后才能执行
        return new PromiseImpl<R>(fulfilledQueue, rejectedQueue, executor, (resolve, reject) -> {
            if (result instanceof Exception) {
                reject.accept((Exception) result);
            } else {
                resolve.accept((R) result);
            }
        });
    }

    @Override
    public <R> Promise<R> onFulfilled(Function<T, R> fulfilled) {
        return then(fulfilled, null);
    }

    @Override
    public <R> Promise<R> onRejected(Function<Exception, R> rejected) {
        return then(null, rejected);
    }


    private static class Null implements Function {
        @Override
        public Object apply(Object o) throws Exception {
            return null;
        }
    }
}
